package com.example.hadoop.mapreduce.dbscan;

import com.example.hadoop.others.Helper;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;

public class DbscanInitMapper extends Mapper<LongWritable, Text, IntWritable, Text> {
    private FileSystem fileSystem;
    private List<Integer> neighbors;
    double eps;
    int minPoints;
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        neighbors = new ArrayList<>();
        String[] temp = DbscanUtils.readParameters(fileSystem, "/cluster/dbscan/params.txt");
        eps = DbscanUtils.getEps(temp);
        minPoints = DbscanUtils.getMinPoints(temp);
        try {
            fileSystem = Helper.getFileSystem();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
        super.setup(context);
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String flag = null;
        String[] temp = value.toString().split(",");
        int index, pIndex;
        double[] data = new double[2];
        double[] pData = new double[2];
        index = Integer.parseInt(temp[0]);
        data[0] = Double.parseDouble(temp[1]);
        data[1] = Double.parseDouble(temp[2]);
        FSDataInputStream inputStream = fileSystem.open(new Path("/cluster/data.txt"));
        BufferedReader bufferedInputStream = new BufferedReader(new InputStreamReader(inputStream));
        String line = null;
        double distance = 0.0;
        while ((line= bufferedInputStream.readLine()) != null) {
            temp = line.split(",");
            pData[0] = Double.parseDouble(temp[1]);
            pData[1] = Double.parseDouble(temp[2]);
            distance = Helper.calDistance(data, pData);
            if (distance < eps) {
                pIndex = Integer.parseInt(temp[0]);
                neighbors.add(pIndex);
            }
        }
        if (minPoints <= neighbors.size()) {
            flag = "core";
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        super.cleanup(context);
    }

    @Override
    public void run(Context context) throws IOException, InterruptedException {
        super.run(context);
    }
}
